<?php
declare (strict_types=1);

namespace app\command;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Cache;

class Idempotent extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('idempotent')
            ->setDescription('the idempotent command');
    }

    protected function execute(Input $input, Output $output)
    {
        // 建立连接
        $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'zq', '123456', '/');
        // 获取信道
        $channel = $connection->channel();
        // 声明交换机
        $channel->exchange_declare('dlx_exchange', 'direct', false, false, false);
        $channel->queue_declare('dlx_queue', false, true, false, false);
        $channel->queue_bind('dlx_queue', 'dlx_exchange', 'dlx_routing_key');

        // 创建重试队列
        $channel->exchange_declare('delay_exchange', 'direct', false, false, false);
        $args = new AMQPTable();
        // 消息过期方式：设置queue.normal队列中的消息5s之后过期
        $args->set('x-message-ttl', 5000);
        // 设置队列最大长度方式： x-max-length
//        $args->set('x-max-length', 1);
        $args->set('x-dead-letter-exchange', 'dlx_exchange');
        $args->set('x-dead-letter-routing-key', 'dlx_routing_key');
        $channel->queue_declare('delay_queue', false, true, false, false, false, $args);
        $channel->queue_bind('delay_queue', 'delay_exchange', 'delay_routing_key');

        // 消息消费
        $channel->basic_consume('dlx_queue', '', false, false, false, false, function ($msg) use ($output) {
            $msg_headers = $msg->get('application_headers')->getNativeData();
            $corr_id = $msg_headers['correlation_id'];

            // 判断是否已经消费过
            if (Cache::get($corr_id) === null) {
//                $body = iconv("UTF-8", "GB2312//IGNORE", "该消息已消费，不再消费");
                $output->writeln(date("h:i:s") . '该消息已消费，不再消费' . PHP_EOL);
                // 确认消息处理完成
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
                return;
            }

//            $body = iconv("UTF-8", "GB2312//IGNORE", $msg->body);
            $output->writeln(date("h:i:s") . " Received " . $msg->body . PHP_EOL);

            // 确认消息处理完成
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            Cache::delete($corr_id);
        });

        while (count($channel->callbacks)) {
            $channel->wait();
        }

        // 关闭信道
        $channel->close();
        // 关闭连接
        $connection->close();
    }
}
